package factory

import (
	"context"
	"errors"
	"fmt"
	"github.com/go-rabbitmq-pool/pool"
	"github.com/streadway/amqp"
	"sync"
	"time"
)

var (
	HOST        = "amqp://guest:guest@localhost:5672/"
	InitTialCap = 4
	MaxIdle     = 8
	MaxCap      = 12
	IdleTimeOut = 15
)

type RabbitFactory struct {
	mx       sync.Mutex
	connPool pool.Pool
}

func (rabbitFactory *RabbitFactory) NewConnPool(host string, initialCap, maxIdle, maxCap, idleTimeout int) error {
	var err error
	connPool, err := initPool(host, initialCap, maxIdle, maxCap, idleTimeout)
	if err != nil {
		return err
	}
	rabbitFactory.connPool = connPool
	return nil
}

func initPool(host string, initialCap, maxIdle, maxCap, IdleTimeout int) (pool.Pool, error) {

	//创建一个连接池： 初始化2，最大连接5，空闲连接数是4
	poolConfig := &pool.Config{
		InitialCap: initialCap,
		MaxIdle:    maxIdle,
		MaxCap:     maxCap,
		Factory: func() (interface{}, error) {
			connection, err := amqp.Dial(host)
			if err != nil {
				fmt.Println(err)
			}
			return connection, err
		},
		Close: func(v interface{}) error {
			connection := v.(*amqp.Connection)
			return connection.Close()

		},
		Ping: func(v interface{}) error {
			isClosed := v.(*amqp.Connection).IsClosed()
			if isClosed {
				return errors.New("connection is Closed")
			}
			return nil
		},
		//连接最大空闲时间，超过该时间的连接 将会关闭，可避免空闲时连接EOF，自动失效的问题
		IdleTimeout: 15 * time.Second,
	}
	p, err := pool.NewChannelPool(poolConfig)
	if err != nil {
		fmt.Println("err=", err)
	}
	return p, err

}

// ExchangeDeclare 创建延时对接交换机
func (rabbitFactory *RabbitFactory) ExchangeDeclareDelayed(name string) (err error) {
	var m = make(map[string]interface{})
	m["x-delayed-type"] = "direct"
	if err != nil {
		return err
	}
	conn, err := rabbitFactory.getConnection()
	defer rabbitFactory.connPool.Put(conn)
	if err != nil {
		return err
	}
	ch, err := conn.Channel()
	defer ch.Close()
	if err != nil {
		return err
	}
	if err != nil {
		return err
	}
	return ch.ExchangeDeclare(name, "x-delayed-message", true, false, false, false, m)

}

func (rabbitFactory *RabbitFactory) getConnection() (*amqp.Connection, error) {
	defer rabbitFactory.mx.Unlock()
	rabbitFactory.mx.Lock()
	conn, err := rabbitFactory.connPool.Get()
	if err != nil {
		return nil, err
	}
	if _, ok := conn.(*amqp.Connection); !ok {
		return nil, errors.New("pool get err")
	}
	if err != nil {
		return nil, errors.New("pool get err")

	}
	connection := conn.(*amqp.Connection)
	return connection, nil

}

// ExchangeDeclare 创建交换机.
func (rabbitFactory *RabbitFactory) ExchangeDeclareArgs(name string, kind string, args map[string]interface{}) (err error) {
	conn, err := rabbitFactory.getConnection()
	if err != nil {
		return err
	}
	ch, err := conn.Channel()
	if err != nil {
		return err
	}
	return ch.ExchangeDeclare(name, kind, true, false, false, false, args)
}

// ExchangeDeclare 创建交换机.
func (rabbitFactory *RabbitFactory) ExchangeDeclare(name string, kind string) (err error) {

	return rabbitFactory.ExchangeDeclareArgs(name, kind, nil)
}

// Publish 发布消息.
func (rabbitFactory *RabbitFactory) Publish(exchange, key string, body []byte) error {

	return rabbitFactory.PublishArgs(exchange, key, body, nil)
}

func (rabbitFactory *RabbitFactory) PublishDelayed(exchange, key string, body []byte, duration time.Duration) error {
	var m = make(map[string]interface{})
	m["x-delay"] = duration.Milliseconds()
	return rabbitFactory.PublishArgs(exchange, key, body, m)
}

func (rabbitFactory *RabbitFactory) PublishArgs(exchange, key string, body []byte, args map[string]interface{}) error {
	conn, err := rabbitFactory.getConnection()
	defer rabbitFactory.connPool.Put(conn)
	if err != nil {
		return err
	}
	ch, err := conn.Channel()
	defer ch.Close()
	if err != nil {
		return err
	}

	if err != nil {
		return err
	}
	err = ch.Publish(
		exchange, // exchange 交换机名称 简单模式下交换机会使用默认的""
		key,      // routing key 路由名称
		false,    // mandatory
		false,    // immediate
		amqp.Publishing{ //发送消息数据
			ContentType: "text/plain",
			Body:        body,
			Headers:     args,
		})
	return err
}

// QueueDeclare 创建队列.
func (rabbitFactory *RabbitFactory) QueueDeclare(name string) (err error) {
	conn, err := rabbitFactory.getConnection()
	defer rabbitFactory.connPool.Put(conn)
	if err != nil {
		return err
	}
	ch, err := conn.Channel()
	defer ch.Close()
	if err != nil {
		return err
	}
	_, err = ch.QueueDeclare(name, true, false, false, false, nil)
	return err
}

// QueueBind 绑定队列.
func (rabbitFactory *RabbitFactory) QueueBind(name, key, exchange string) (err error) {
	conn, err := rabbitFactory.getConnection()
	defer rabbitFactory.connPool.Put(conn)
	if err != nil {
		return err
	}
	ch, err := conn.Channel()
	defer ch.Close()
	if err != nil {
		return err
	}
	if err != nil {
		return err
	}
	return ch.QueueBind(name, key, exchange, false, nil)
}

// NewConsumer 实例化一个消费者, 会单独用一个channel.
func (rabbitFactory *RabbitFactory) NewConsumer(ctx context.Context, queue string, handler func([]byte) error) error {
	conn, err := rabbitFactory.getConnection()
	defer rabbitFactory.connPool.Put(conn)
	if err != nil {
		return err
	}
	ch, err := conn.Channel()
	if err != nil {
		return err
	}
	deliveries, err := ch.Consume(queue, "", false, false, false, false, nil)
	if err != nil {
		return fmt.Errorf("consume err: %v, queue: %s", err, queue)
	}
	for msg := range deliveries {
		select {
		case <-ctx.Done():
			_ = msg.Reject(true)
			return fmt.Errorf("context cancel")
		default:
		}
		err = handler(msg.Body)
		if err != nil {
			_ = msg.Reject(true)
			continue
		}
		_ = msg.Ack(false)
	}
	return nil
}
